pgsql.c (18500B)
1 /*
2 * Copyright (c) 2014-2022 Joris Vink <joris@coders.se>
3 *
4 * Permission to use, copy, modify, and distribute this software for any
5 * purpose with or without fee is hereby granted, provided that the above
6 * copyright notice and this permission notice appear in all copies.
7 *
8 * THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
9 * WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
10 * MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
11 * ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
12 * WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
13 * ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
14 * OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
15 */
16
17 #include <sys/param.h>
18 #include <sys/queue.h>
19
20 #include <libpq-fe.h>
21 #include <pg_config.h>
22
23 #include "kore.h"
24
25 #if !defined(KORE_NO_HTTP)
26 #include "http.h"
27 #endif
28
29 #include "pgsql.h"
30
31 #if defined(__linux__)
32 #include "seccomp.h"
33
34 static struct sock_filter filter_pgsql[] = {
35 /* Allow us to create sockets and call connect. */
36 KORE_SYSCALL_ALLOW(connect),
37 KORE_SYSCALL_ALLOW_ARG(socket, 0, AF_INET),
38 KORE_SYSCALL_ALLOW_ARG(socket, 0, AF_INET6),
39 KORE_SYSCALL_ALLOW_ARG(socket, 0, AF_UNIX),
40
41 /* Requires these calls. */
42 KORE_SYSCALL_ALLOW(getsockopt),
43 KORE_SYSCALL_ALLOW(getsockname),
44 };
45 #endif
46
47 struct pgsql_wait {
48 struct kore_pgsql *pgsql;
49 TAILQ_ENTRY(pgsql_wait) list;
50 };
51
52 struct pgsql_job {
53 struct kore_pgsql *pgsql;
54 TAILQ_ENTRY(pgsql_job) list;
55 };
56
57 #define PGSQL_CONN_MAX 2
58 #define PGSQL_CONN_FREE 0x01
59 #define PGSQL_LIST_INSERTED 0x0100
60 #define PGSQL_QUEUE_LIMIT 1000
61
62 static void pgsql_queue_wakeup(void);
63 static void pgsql_cancel(struct kore_pgsql *);
64 static void pgsql_set_error(struct kore_pgsql *, const char *);
65 static void pgsql_queue_add(struct kore_pgsql *);
66 static void pgsql_queue_remove(struct kore_pgsql *);
67 static void pgsql_conn_release(struct kore_pgsql *);
68 static void pgsql_conn_cleanup(struct pgsql_conn *);
69 static void pgsql_read_result(struct kore_pgsql *);
70 static void pgsql_schedule(struct kore_pgsql *);
71
72 static struct pgsql_conn *pgsql_conn_create(struct kore_pgsql *,
73 struct pgsql_db *);
74 static struct pgsql_conn *pgsql_conn_next(struct kore_pgsql *,
75 struct pgsql_db *);
76
77 static struct kore_pool pgsql_job_pool;
78 static struct kore_pool pgsql_wait_pool;
79 static TAILQ_HEAD(, pgsql_conn) pgsql_conn_free;
80 static TAILQ_HEAD(, pgsql_wait) pgsql_wait_queue;
81 static LIST_HEAD(, pgsql_db) pgsql_db_conn_strings;
82
83 u_int32_t pgsql_queue_count = 0;
84 u_int16_t pgsql_conn_max = PGSQL_CONN_MAX;
85 u_int32_t pgsql_queue_limit = PGSQL_QUEUE_LIMIT;
86
87 void
88 kore_pgsql_sys_init(void)
89 {
90 TAILQ_INIT(&pgsql_conn_free);
91 TAILQ_INIT(&pgsql_wait_queue);
92 LIST_INIT(&pgsql_db_conn_strings);
93
94 kore_pool_init(&pgsql_job_pool, "pgsql_job_pool",
95 sizeof(struct pgsql_job), 100);
96 kore_pool_init(&pgsql_wait_pool, "pgsql_wait_pool",
97 sizeof(struct pgsql_wait), pgsql_queue_limit);
98
99 #if defined(__linux__)
100 kore_seccomp_filter("pgsql", filter_pgsql,
101 KORE_FILTER_LEN(filter_pgsql));
102 #endif
103 }
104
105 void
106 kore_pgsql_sys_cleanup(void)
107 {
108 struct pgsql_conn *conn, *next;
109
110 kore_pool_cleanup(&pgsql_job_pool);
111 kore_pool_cleanup(&pgsql_wait_pool);
112
113 for (conn = TAILQ_FIRST(&pgsql_conn_free); conn != NULL; conn = next) {
114 next = TAILQ_NEXT(conn, list);
115 pgsql_conn_cleanup(conn);
116 }
117 }
118
119 void
120 kore_pgsql_init(struct kore_pgsql *pgsql)
121 {
122 memset(pgsql, 0, sizeof(*pgsql));
123 pgsql->state = KORE_PGSQL_STATE_INIT;
124 }
125
126 int
127 kore_pgsql_setup(struct kore_pgsql *pgsql, const char *dbname, int flags)
128 {
129 struct pgsql_db *db;
130
131 if ((flags & KORE_PGSQL_ASYNC) && (flags & KORE_PGSQL_SYNC)) {
132 pgsql_set_error(pgsql, "invalid query init parameters");
133 return (KORE_RESULT_ERROR);
134 }
135
136 if (flags & KORE_PGSQL_ASYNC) {
137 if (pgsql->req == NULL && pgsql->cb == NULL) {
138 pgsql_set_error(pgsql, "nothing was bound");
139 return (KORE_RESULT_ERROR);
140 }
141 }
142
143 db = NULL;
144 pgsql->flags |= flags;
145
146 LIST_FOREACH(db, &pgsql_db_conn_strings, rlist) {
147 if (!strcmp(db->name, dbname))
148 break;
149 }
150
151 if (db == NULL) {
152 pgsql_set_error(pgsql, "no database found");
153 return (KORE_RESULT_ERROR);
154 }
155
156 if ((pgsql->conn = pgsql_conn_next(pgsql, db)) == NULL)
157 return (KORE_RESULT_ERROR);
158
159 if (pgsql->flags & KORE_PGSQL_ASYNC) {
160 pgsql->conn->job = kore_pool_get(&pgsql_job_pool);
161 pgsql->conn->job->pgsql = pgsql;
162 }
163
164 return (KORE_RESULT_OK);
165 }
166
167 #if !defined(KORE_NO_HTTP)
168 void
169 kore_pgsql_bind_request(struct kore_pgsql *pgsql, struct http_request *req)
170 {
171 if (pgsql->req != NULL || pgsql->cb != NULL)
172 fatal("kore_pgsql_bind_request: already bound");
173
174 pgsql->req = req;
175 pgsql->flags |= PGSQL_LIST_INSERTED;
176
177 LIST_INSERT_HEAD(&(req->pgsqls), pgsql, rlist);
178 }
179 #endif
180
181 void
182 kore_pgsql_bind_callback(struct kore_pgsql *pgsql,
183 void (*cb)(struct kore_pgsql *, void *), void *arg)
184 {
185 if (pgsql->req != NULL)
186 fatal("kore_pgsql_bind_callback: already bound");
187
188 if (pgsql->cb != NULL)
189 fatal("kore_pgsql_bind_callback: already bound");
190
191 pgsql->cb = cb;
192 pgsql->arg = arg;
193 }
194
195 int
196 kore_pgsql_query(struct kore_pgsql *pgsql, const void *query)
197 {
198 if (pgsql->conn == NULL) {
199 pgsql_set_error(pgsql, "no connection was set before query");
200 return (KORE_RESULT_ERROR);
201 }
202
203 if (pgsql->flags & KORE_PGSQL_SYNC) {
204 pgsql->result = PQexec(pgsql->conn->db, query);
205 if ((PQresultStatus(pgsql->result) != PGRES_TUPLES_OK) &&
206 (PQresultStatus(pgsql->result) != PGRES_COMMAND_OK)) {
207 pgsql_set_error(pgsql, PQerrorMessage(pgsql->conn->db));
208 return (KORE_RESULT_ERROR);
209 }
210
211 pgsql->state = KORE_PGSQL_STATE_DONE;
212 } else {
213 if (!PQsendQuery(pgsql->conn->db, query)) {
214 pgsql_set_error(pgsql, PQerrorMessage(pgsql->conn->db));
215 return (KORE_RESULT_ERROR);
216 }
217
218 pgsql_schedule(pgsql);
219 }
220
221 return (KORE_RESULT_OK);
222 }
223
224 int
225 kore_pgsql_v_query_params(struct kore_pgsql *pgsql,
226 const void *query, int binary, int count, va_list args)
227 {
228 int i;
229 const char **values;
230 int *lengths, *formats, ret;
231
232 if (pgsql->conn == NULL) {
233 pgsql_set_error(pgsql, "no connection was set before query");
234 return (KORE_RESULT_ERROR);
235 }
236
237 if (count > 0) {
238 lengths = kore_calloc(count, sizeof(int));
239 formats = kore_calloc(count, sizeof(int));
240 values = kore_calloc(count, sizeof(char *));
241
242 for (i = 0; i < count; i++) {
243 values[i] = va_arg(args, void *);
244 lengths[i] = va_arg(args, int);
245 formats[i] = va_arg(args, int);
246 }
247 } else {
248 lengths = NULL;
249 formats = NULL;
250 values = NULL;
251 }
252
253 ret = kore_pgsql_query_param_fields(pgsql, query, binary, count,
254 values, lengths, formats);
255
256 kore_free(values);
257 kore_free(lengths);
258 kore_free(formats);
259
260 return (ret);
261 }
262
263 int
264 kore_pgsql_query_param_fields(struct kore_pgsql *pgsql, const void *query,
265 int binary, int count, const char **values, int *lengths, int *formats)
266 {
267 if (pgsql->conn == NULL) {
268 pgsql_set_error(pgsql, "no connection was set before query");
269 return (KORE_RESULT_ERROR);
270 }
271
272 if (pgsql->flags & KORE_PGSQL_SYNC) {
273 pgsql->result = PQexecParams(pgsql->conn->db, query, count,
274 NULL, (const char * const *)values, lengths, formats,
275 binary);
276
277 if ((PQresultStatus(pgsql->result) != PGRES_TUPLES_OK) &&
278 (PQresultStatus(pgsql->result) != PGRES_COMMAND_OK)) {
279 pgsql_set_error(pgsql, PQerrorMessage(pgsql->conn->db));
280 return (KORE_RESULT_ERROR);
281 }
282
283 pgsql->state = KORE_PGSQL_STATE_DONE;
284 } else {
285 if (!PQsendQueryParams(pgsql->conn->db, query, count, NULL,
286 (const char * const *)values, lengths, formats, binary)) {
287 pgsql_set_error(pgsql, PQerrorMessage(pgsql->conn->db));
288 return (KORE_RESULT_ERROR);
289 }
290
291 pgsql_schedule(pgsql);
292 }
293
294 return (KORE_RESULT_OK);
295 }
296
297 int
298 kore_pgsql_query_params(struct kore_pgsql *pgsql,
299 const void *query, int binary, int count, ...)
300 {
301 int ret;
302 va_list args;
303
304 va_start(args, count);
305 ret = kore_pgsql_v_query_params(pgsql, query, binary, count, args);
306 va_end(args);
307
308 return (ret);
309 }
310
311 int
312 kore_pgsql_register(const char *dbname, const char *connstring)
313 {
314 struct pgsql_db *pgsqldb;
315
316 LIST_FOREACH(pgsqldb, &pgsql_db_conn_strings, rlist) {
317 if (!strcmp(pgsqldb->name, dbname))
318 return (KORE_RESULT_ERROR);
319 }
320
321 pgsqldb = kore_malloc(sizeof(*pgsqldb));
322 pgsqldb->name = kore_strdup(dbname);
323 pgsqldb->conn_count = 0;
324 pgsqldb->conn_max = pgsql_conn_max;
325 pgsqldb->conn_string = kore_strdup(connstring);
326 LIST_INSERT_HEAD(&pgsql_db_conn_strings, pgsqldb, rlist);
327
328 return (KORE_RESULT_OK);
329 }
330
331 void
332 kore_pgsql_handle(void *c, int err)
333 {
334 struct kore_pgsql *pgsql;
335 struct pgsql_conn *conn = (struct pgsql_conn *)c;
336
337 if (err) {
338 pgsql_conn_cleanup(conn);
339 return;
340 }
341
342 if (!(conn->evt.flags & KORE_EVENT_READ))
343 fatal("%s: read event not set", __func__);
344
345 pgsql = conn->job->pgsql;
346
347 pgsql_read_result(pgsql);
348
349 if (pgsql->state == KORE_PGSQL_STATE_WAIT) {
350 #if !defined(KORE_NO_HTTP)
351 if (pgsql->req != NULL)
352 http_request_sleep(pgsql->req);
353 #endif
354 if (pgsql->cb != NULL)
355 pgsql->cb(pgsql, pgsql->arg);
356 } else {
357 #if !defined(KORE_NO_HTTP)
358 if (pgsql->req != NULL)
359 http_request_wakeup(pgsql->req);
360 #endif
361 if (pgsql->cb != NULL)
362 pgsql->cb(pgsql, pgsql->arg);
363 }
364 }
365
366 void
367 kore_pgsql_continue(struct kore_pgsql *pgsql)
368 {
369 if (pgsql->error) {
370 kore_free(pgsql->error);
371 pgsql->error = NULL;
372 }
373
374 if (pgsql->result) {
375 PQclear(pgsql->result);
376 pgsql->result = NULL;
377 }
378
379 switch (pgsql->state) {
380 case KORE_PGSQL_STATE_INIT:
381 case KORE_PGSQL_STATE_WAIT:
382 break;
383 case KORE_PGSQL_STATE_DONE:
384 #if !defined(KORE_NO_HTTP)
385 if (pgsql->req != NULL)
386 http_request_wakeup(pgsql->req);
387 #endif
388 pgsql_conn_release(pgsql);
389 break;
390 case KORE_PGSQL_STATE_ERROR:
391 case KORE_PGSQL_STATE_RESULT:
392 case KORE_PGSQL_STATE_NOTIFY:
393 kore_pgsql_handle(pgsql->conn, 0);
394 break;
395 default:
396 fatal("unknown pgsql state %d", pgsql->state);
397 }
398 }
399
400 void
401 kore_pgsql_cleanup(struct kore_pgsql *pgsql)
402 {
403 pgsql_queue_remove(pgsql);
404
405 if (pgsql->result != NULL)
406 PQclear(pgsql->result);
407
408 if (pgsql->error != NULL)
409 kore_free(pgsql->error);
410
411 if (pgsql->conn != NULL)
412 pgsql_conn_release(pgsql);
413
414 pgsql->result = NULL;
415 pgsql->error = NULL;
416 pgsql->conn = NULL;
417
418 if (pgsql->flags & PGSQL_LIST_INSERTED) {
419 LIST_REMOVE(pgsql, rlist);
420 pgsql->flags &= ~PGSQL_LIST_INSERTED;
421 }
422 }
423
424 void
425 kore_pgsql_logerror(struct kore_pgsql *pgsql)
426 {
427 kore_log(LOG_NOTICE, "pgsql error: %s",
428 (pgsql->error) ? pgsql->error : "unknown");
429 }
430
431 int
432 kore_pgsql_ntuples(struct kore_pgsql *pgsql)
433 {
434 return (PQntuples(pgsql->result));
435 }
436
437 int
438 kore_pgsql_nfields(struct kore_pgsql *pgsql)
439 {
440 return (PQnfields(pgsql->result));
441 }
442
443 int
444 kore_pgsql_getlength(struct kore_pgsql *pgsql, int row, int col)
445 {
446 return (PQgetlength(pgsql->result, row, col));
447 }
448
449 char *
450 kore_pgsql_fieldname(struct kore_pgsql *pgsql, int field)
451 {
452 return (PQfname(pgsql->result, field));
453 }
454
455 char *
456 kore_pgsql_getvalue(struct kore_pgsql *pgsql, int row, int col)
457 {
458 return (PQgetvalue(pgsql->result, row, col));
459 }
460
461 int
462 kore_pgsql_column_binary(struct kore_pgsql *pgsql, int col)
463 {
464 return (PQfformat(pgsql->result, col));
465 }
466
467 static struct pgsql_conn *
468 pgsql_conn_next(struct kore_pgsql *pgsql, struct pgsql_db *db)
469 {
470 PGTransactionStatusType state;
471 struct pgsql_conn *conn;
472 struct kore_pgsql rollback;
473
474 rescan:
475 conn = NULL;
476
477 TAILQ_FOREACH(conn, &pgsql_conn_free, list) {
478 if (!(conn->flags & PGSQL_CONN_FREE))
479 fatal("got a pgsql connection that was not free?");
480 if (!strcmp(conn->name, db->name))
481 break;
482 }
483
484 if (conn != NULL) {
485 state = PQtransactionStatus(conn->db);
486 if (state == PQTRANS_INERROR) {
487 conn->flags &= ~PGSQL_CONN_FREE;
488 TAILQ_REMOVE(&pgsql_conn_free, conn, list);
489
490 kore_pgsql_init(&rollback);
491 rollback.conn = conn;
492 rollback.flags = KORE_PGSQL_SYNC;
493
494 if (!kore_pgsql_query(&rollback, "ROLLBACK")) {
495 kore_pgsql_logerror(&rollback);
496 kore_pgsql_cleanup(&rollback);
497 pgsql_conn_cleanup(conn);
498 } else {
499 kore_pgsql_cleanup(&rollback);
500 }
501
502 goto rescan;
503 }
504 }
505
506 if (conn == NULL) {
507 if (db->conn_max != 0 &&
508 db->conn_count >= db->conn_max) {
509 if ((pgsql->flags & KORE_PGSQL_ASYNC) &&
510 pgsql_queue_count < pgsql_queue_limit) {
511 pgsql_queue_add(pgsql);
512 } else {
513 pgsql_set_error(pgsql,
514 "no available connection");
515 }
516
517 return (NULL);
518 }
519
520 if ((conn = pgsql_conn_create(pgsql, db)) == NULL)
521 return (NULL);
522 }
523
524 conn->flags &= ~PGSQL_CONN_FREE;
525 TAILQ_REMOVE(&pgsql_conn_free, conn, list);
526
527 return (conn);
528 }
529
530 static void
531 pgsql_set_error(struct kore_pgsql *pgsql, const char *msg)
532 {
533 if (pgsql->error != NULL)
534 kore_free(pgsql->error);
535
536 pgsql->error = kore_strdup(msg);
537 pgsql->state = KORE_PGSQL_STATE_ERROR;
538 }
539
540 static void
541 pgsql_schedule(struct kore_pgsql *pgsql)
542 {
543 int fd;
544
545 fd = PQsocket(pgsql->conn->db);
546 if (fd < 0)
547 fatal("PQsocket returned < 0 fd on open connection");
548
549 kore_platform_schedule_read(fd, pgsql->conn);
550 pgsql->state = KORE_PGSQL_STATE_WAIT;
551 pgsql->flags |= KORE_PGSQL_SCHEDULED;
552
553 #if !defined(KORE_NO_HTTP)
554 if (pgsql->req != NULL)
555 http_request_sleep(pgsql->req);
556 #endif
557 if (pgsql->cb != NULL)
558 pgsql->cb(pgsql, pgsql->arg);
559 }
560
561 static void
562 pgsql_queue_add(struct kore_pgsql *pgsql)
563 {
564 struct pgsql_wait *pgw;
565
566 #if !defined(KORE_NO_HTTP)
567 if (pgsql->req != NULL)
568 http_request_sleep(pgsql->req);
569 #endif
570
571 pgw = kore_pool_get(&pgsql_wait_pool);
572 pgw->pgsql = pgsql;
573
574 pgsql_queue_count++;
575 TAILQ_INSERT_TAIL(&pgsql_wait_queue, pgw, list);
576 }
577
578 static void
579 pgsql_queue_remove(struct kore_pgsql *pgsql)
580 {
581 struct pgsql_wait *pgw, *next;
582
583 for (pgw = TAILQ_FIRST(&pgsql_wait_queue); pgw != NULL; pgw = next) {
584 next = TAILQ_NEXT(pgw, list);
585 if (pgw->pgsql != pgsql)
586 continue;
587
588 pgsql_queue_count--;
589 TAILQ_REMOVE(&pgsql_wait_queue, pgw, list);
590 kore_pool_put(&pgsql_wait_pool, pgw);
591 return;
592 }
593 }
594
595 static void
596 pgsql_queue_wakeup(void)
597 {
598 struct pgsql_wait *pgw, *next;
599
600 for (pgw = TAILQ_FIRST(&pgsql_wait_queue); pgw != NULL; pgw = next) {
601 next = TAILQ_NEXT(pgw, list);
602
603 #if !defined(KORE_NO_HTTP)
604 if (pgw->pgsql->req != NULL) {
605 if (pgw->pgsql->req->flags & HTTP_REQUEST_DELETE) {
606 pgsql_queue_count--;
607 TAILQ_REMOVE(&pgsql_wait_queue, pgw, list);
608 kore_pool_put(&pgsql_wait_pool, pgw);
609 continue;
610 }
611
612 http_request_wakeup(pgw->pgsql->req);
613 }
614 #endif
615 if (pgw->pgsql->cb != NULL)
616 pgw->pgsql->cb(pgw->pgsql, pgw->pgsql->arg);
617
618 pgsql_queue_count--;
619 TAILQ_REMOVE(&pgsql_wait_queue, pgw, list);
620 kore_pool_put(&pgsql_wait_pool, pgw);
621 return;
622 }
623 }
624
625 static struct pgsql_conn *
626 pgsql_conn_create(struct kore_pgsql *pgsql, struct pgsql_db *db)
627 {
628 struct pgsql_conn *conn;
629
630 if (db == NULL || db->conn_string == NULL)
631 fatal("pgsql_conn_create: no connection string");
632
633 db->conn_count++;
634
635 conn = kore_calloc(1, sizeof(*conn));
636 conn->job = NULL;
637 conn->flags = PGSQL_CONN_FREE;
638 conn->name = kore_strdup(db->name);
639 TAILQ_INSERT_TAIL(&pgsql_conn_free, conn, list);
640
641 conn->evt.type = KORE_TYPE_PGSQL_CONN;
642 conn->evt.handle = kore_pgsql_handle;
643
644 conn->db = PQconnectdb(db->conn_string);
645 if (conn->db == NULL || (PQstatus(conn->db) != CONNECTION_OK)) {
646 pgsql_set_error(pgsql, PQerrorMessage(conn->db));
647 pgsql_conn_cleanup(conn);
648 return (NULL);
649 }
650
651 return (conn);
652 }
653
654 static void
655 pgsql_conn_release(struct kore_pgsql *pgsql)
656 {
657 int fd;
658 PGresult *result;
659
660 if (pgsql->conn == NULL)
661 return;
662
663 /* Async query cleanup */
664 if (pgsql->flags & KORE_PGSQL_ASYNC) {
665 if (pgsql->flags & KORE_PGSQL_SCHEDULED) {
666 fd = PQsocket(pgsql->conn->db);
667 kore_platform_disable_read(fd);
668
669 if (pgsql->state != KORE_PGSQL_STATE_DONE)
670 pgsql_cancel(pgsql);
671 }
672 kore_pool_put(&pgsql_job_pool, pgsql->conn->job);
673 }
674
675 /* Drain just in case. */
676 while ((result = PQgetResult(pgsql->conn->db)) != NULL)
677 PQclear(result);
678
679 pgsql->conn->job = NULL;
680 pgsql->conn->flags |= PGSQL_CONN_FREE;
681 TAILQ_INSERT_TAIL(&pgsql_conn_free, pgsql->conn, list);
682
683 pgsql->conn = NULL;
684 pgsql->state = KORE_PGSQL_STATE_COMPLETE;
685
686 if (pgsql->cb != NULL)
687 pgsql->cb(pgsql, pgsql->arg);
688
689 pgsql_queue_wakeup();
690 }
691
692 static void
693 pgsql_conn_cleanup(struct pgsql_conn *conn)
694 {
695 struct kore_pgsql *pgsql;
696 struct pgsql_db *pgsqldb;
697
698 if (conn->flags & PGSQL_CONN_FREE)
699 TAILQ_REMOVE(&pgsql_conn_free, conn, list);
700
701 if (conn->job) {
702 pgsql = conn->job->pgsql;
703 #if !defined(KORE_NO_HTTP)
704 if (pgsql->req != NULL)
705 http_request_wakeup(pgsql->req);
706 #endif
707 pgsql->conn = NULL;
708 pgsql_set_error(pgsql, PQerrorMessage(conn->db));
709
710 kore_pool_put(&pgsql_job_pool, conn->job);
711 conn->job = NULL;
712 }
713
714 if (conn->db != NULL)
715 PQfinish(conn->db);
716
717 LIST_FOREACH(pgsqldb, &pgsql_db_conn_strings, rlist) {
718 if (!strcmp(pgsqldb->name, conn->name)) {
719 pgsqldb->conn_count--;
720 break;
721 }
722 }
723
724 kore_free(conn->name);
725 kore_free(conn);
726 }
727
728 static void
729 pgsql_read_result(struct kore_pgsql *pgsql)
730 {
731 struct pgsql_conn *conn;
732 PGnotify *notify;
733 int saved_errno;
734
735 conn = pgsql->conn;
736
737 for (;;) {
738 if (!PQconsumeInput(conn->db)) {
739 pgsql->state = KORE_PGSQL_STATE_ERROR;
740 pgsql->error = kore_strdup(PQerrorMessage(conn->db));
741 return;
742 }
743
744 saved_errno = errno;
745
746 if (PQisBusy(conn->db)) {
747 if (saved_errno != EAGAIN && saved_errno != EWOULDBLOCK)
748 continue;
749 pgsql->state = KORE_PGSQL_STATE_WAIT;
750 conn->evt.flags &= ~KORE_EVENT_READ;
751 return;
752 }
753
754 break;
755 }
756
757 while ((notify = PQnotifies(conn->db)) != NULL) {
758 pgsql->state = KORE_PGSQL_STATE_NOTIFY;
759 pgsql->notify.extra = notify->extra;
760 pgsql->notify.channel = notify->relname;
761
762 if (pgsql->cb != NULL)
763 pgsql->cb(pgsql, pgsql->arg);
764
765 PQfreemem(notify);
766 }
767
768 pgsql->result = PQgetResult(conn->db);
769 if (pgsql->result == NULL) {
770 pgsql->state = KORE_PGSQL_STATE_DONE;
771 return;
772 }
773
774 switch (PQresultStatus(pgsql->result)) {
775 #if PG_VERSION_NUM >= 140000
776 case PGRES_PIPELINE_SYNC:
777 case PGRES_PIPELINE_ABORTED:
778 #endif
779 case PGRES_COPY_OUT:
780 case PGRES_COPY_IN:
781 case PGRES_NONFATAL_ERROR:
782 case PGRES_COPY_BOTH:
783 break;
784 case PGRES_COMMAND_OK:
785 pgsql->state = KORE_PGSQL_STATE_DONE;
786 break;
787 case PGRES_TUPLES_OK:
788 #if PG_VERSION_NUM >= 90200
789 case PGRES_SINGLE_TUPLE:
790 #endif
791 pgsql->state = KORE_PGSQL_STATE_RESULT;
792 break;
793 case PGRES_EMPTY_QUERY:
794 case PGRES_BAD_RESPONSE:
795 case PGRES_FATAL_ERROR:
796 pgsql_set_error(pgsql, PQresultErrorMessage(pgsql->result));
797 break;
798 }
799 }
800
801 static void
802 pgsql_cancel(struct kore_pgsql *pgsql)
803 {
804 PGcancel *cancel;
805 char buf[256];
806
807 if ((cancel = PQgetCancel(pgsql->conn->db)) != NULL) {
808 if (!PQcancel(cancel, buf, sizeof(buf)))
809 kore_log(LOG_ERR, "failed to cancel: %s", buf);
810 PQfreeCancel(cancel);
811 }
812 }