7

#include "evsql_internal.h"

    19

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

    21

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

    22

    int err;

    23

    24

    switch (conn->evsql->type) {

    28

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

    39

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

    41

    42

            if (err) {

    43

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

    44

                    WARNING("conn failed");

    45

                else

    46

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

    47

            }

    48

    49

            break;

    54

    55

    if (!err)

    56

        // assign the query

    57

        conn->query = query;

    58

    59

    return err;

    63

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

    64

 *

    65

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

    81

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

    93

 * XXX:

    94

 * /

    99

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

   102

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

   108

*/

   111

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

   113

static void _evsql_trans_free (struct evsql_trans *trans) {

   114

    // ensure we don't leak anything

   115

    assert(trans->query == NULL);

   116

    assert(trans->conn == NULL);

   118

    // free

   119

    free(trans);

   120

}

   121

   122

/*

   123

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

   124

 *

   125

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

   126

 */

   127

static void _evsql_conn_release (struct evsql_conn *conn) {

   128

    // ensure we don't leak anything

   129

    assert(conn->trans == NULL);

   130

    assert(conn->query == NULL);

   131

   132

    // release the engine

   133

    switch (conn->evsql->type) {

   134

        case EVSQL_EVPQ:

   135

            evpq_release(conn->engine.evpq);

   136

            break;

   137

   138

        default:

   139

            FATAL("evsql->type");

   140

    }

   141

   142

    // remove from list

   143

    LIST_REMOVE(conn, entry);

   144

   145

    // free

   146

    free(conn);

   147

}

   148

   149

/*

   150

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

   151

 */

   152

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

   153

    struct evsql_result_info res; ZINIT(res);

   154

   155

    // set up the result_info

   156

    res.evsql = evsql;

   157

    res.error = 1;

   158

   159

    // finish off the query

   160

    _evsql_query_done(query, &res);

   161

}

   162

   163

/*

   164

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

   165

 * conn, and then free the trans.

   166

 */ 

   167

static void _evsql_trans_fail (struct evsql_trans *trans) {

   168

    if (trans->query) {

   169

        // free the query silently

   170

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

   171

    }

   172

   173

    // tell the user

   174

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

   175

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

   176

   177

    // fail the conn

   178

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

   179

   180

    // free the trans

   181

    _evsql_trans_free(trans);

   182

}

   183

   184

/*

   185

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

   186

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

   187

 */

   188

static void _evsql_conn_fail (struct evsql_conn *conn) {

   189

    if (conn->trans) {

   190

        // let transactions handle their connection failures

   191

        _evsql_trans_fail(conn->trans);

   192

   193

    } else {

   194

        if (conn->query) {

   195

            // fail the in-progress query

   196

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

   199

        // finish off the whole connection

   200

        _evsql_conn_release(conn);

   204

/*

   205

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

   206

 *

   207

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

   208

 *

   209

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

   210

 */

   211

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

   212

    struct evsql_query *query;

   213

    int err;

   214

   215

    // look for waiting queries

   216

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

   217

        // dequeue

   218

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

   219

   220

        if (conn) {

   221

            // try and execute it

   222

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

   223

        }

   225

        // free the command buf

   226

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

   228

        if (err || !conn) {

   229

            if (!conn) {

   230

                // warn when dropping queries

   231

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

   232

            }

   233

   234

            // fail the query

   235

            _evsql_query_fail(evsql, query);

   236

   237

            if (conn) {

   238

                // fail the connection

   239

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

   240

   241

                _evsql_conn_fail(conn); conn = NULL;

   242

            }

   243

   244

        } else {

   245

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

   246

            break;

   247

   248

        }

   249

   250

        // handle the rest of the queue

   251

    }

   252

   253

    // ok

   254

    return;

   257

/*

   258

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

   259

 */

   260

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

   261

    (void) arg;

   263

    assert(res->trans);

   264

   265

    // check for errors

   266

    if (res->error)

   267

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

   268

   269

    // transaction is now ready for use

   270

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

   271

   272

error:

   273

    _evsql_trans_fail(res->trans);

   274

}

   275

   276

/*

   277

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

   278

 */

   279

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

   280

    char trans_sql[EVSQL_QUERY_BEGIN_BUF];

   281

    const char *isolation_level;

   282

    int ret;

   283

   284

    // determine the isolation_level to use

   285

    switch (trans->type) {

   286

        case EVSQL_TRANS_DEFAULT:

   287

            isolation_level = NULL; break;

   288

   289

        case EVSQL_TRANS_SERIALIZABLE:

   290

            isolation_level = "SERIALIZABLE"; break;

   291

   292

        case EVSQL_TRANS_REPEATABLE_READ:

   293

            isolation_level = "REPEATABLE READ"; break;

   294

   295

        case EVSQL_TRANS_READ_COMMITTED:

   296

            isolation_level = "READ COMMITTED"; break;

   297

   298

        case EVSQL_TRANS_READ_UNCOMMITTED:

   299

            isolation_level = "READ UNCOMMITTED"; break;

   300

   301

        default:

   302

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

   303

    }

   304

   305

    // build the trans_sql

   306

    if (isolation_level)

   307

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

   308

    else

   309

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

   310

   311

    // make sure it wasn't truncated

   312

    if (ret >= EVSQL_QUERY_BEGIN_BUF)

   313

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

   314

   315

    // execute the query

   316

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

   317

        ERROR("evsql_query");

   318

   319

    // success

   320

    return;

   321

   322

error:

   323

    // fail the transaction

   324

    _evsql_trans_fail(trans);

   325

}

   326

   327

/*

   328

 * The evpq connection was succesfully established.

   329

 */ 

   330

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

   331

    struct evsql_conn *conn = arg;

   332

   333

    if (conn->trans)

   334

        // notify the transaction

   335

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

   336

   337

    else

   338

        // pump any waiting transactionless queries

   339

        _evsql_pump(conn->evsql, conn);

   340

}

   341

   342

/*

   343

 * Got one result on this evpq connection.

   344

 */

   345

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

   346

    struct evsql_conn *conn = arg;

   347

    struct evsql_query *query = conn->query;

   348

   349

    assert(query != NULL);

   362

/*

   363

 * No more results for this query.

   364

 */

   365

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

   366

    struct evsql_conn *conn = arg;

   367

    struct evsql_query *query = conn->query;

   369

   370

    assert(query != NULL);

   373

    res.evsql = conn->evsql;

   392

    // de-associate the query from the connection

   393

    conn->query = NULL;

   394

   395

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

   396

    if (conn->trans) {

   397

        // we can deassign the trans's query

   398

        conn->trans->query = NULL;

   400

        // then hand the query to the user

   401

        _evsql_query_done(query, &res);

   402

   403

    } else {

   404

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

   405

        _evsql_query_done(query, &res);

   406

   407

        // pump the next one

   408

        _evsql_pump(conn->evsql, conn);

   409

    }

   412

/*

   413

 * The connection failed.

   414

 */

   415

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

   416

    struct evsql_conn *conn = arg;

   418

    // just fail the conn

   419

    _evsql_conn_fail(conn);

   422

/*

   423

 * Our evpq behaviour

   424

 */

   432

/*

   433

 * Allocate the generic evsql context.

   434

 */

   435

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

   443

    evsql->ev_base = ev_base;

   448

    LIST_INIT(&evsql->conn_list);

   449

    TAILQ_INIT(&evsql->query_queue);

   458

/*

   459

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

   460

 */

   461

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

   462

    struct evsql_conn *conn = NULL;

   463

   464

    // allocate

   465

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

   466

        ERROR("calloc");

   467

   468

    // init

   469

    conn->evsql = evsql;

   470

   471

    // connect the engine

   472

    switch (evsql->type) {

   473

        case EVSQL_EVPQ:

   474

            if ((conn->engine.evpq = evpq_connect(evsql->ev_base, evsql->engine_conf.evpq, _evsql_evpq_cb_info, conn)) == NULL)

   475

                goto error;

   476

   477

            break;

   478

   479

        default:

   480

            FATAL("evsql->type");

   481

    }

   482

   483

    // add it to the list

   484

    LIST_INSERT_HEAD(&evsql->conn_list, conn, entry);

   485

   486

    // success

   487

    return conn;

   488

   489

error:

   490

    free(conn);

   491

   492

    return NULL;

   493

}

   494