← Back to branch summary

~psycopg/psycopg/2.0.x

« back to all changes in this revision

Viewing changes to psycopg/cursor_type.c

  • Committer: Federico Di Gregorio
  • Date: 2004-12-10 10:34:57 UTC
  • Revision ID: fog-1b2e9ebeaa5d76bd7e4a9dc5eb779afb4076f236
async fixes and better connection/cursor management.

Show diffs side-by-side

added added

removed removed

Lines of Context:
psycopg/cursor_type.c
249
249
        return NULL;
250
250
    }
251
251
 
 
252
    pthread_mutex_lock(&(self->conn->lock));
252
253
    if (self->conn->async_cursor != NULL
253
254
        && self->conn->async_cursor != (PyObject*)self) {
 
255
        pthread_mutex_unlock(&(self->conn->lock));
254
256
        PyErr_SetString(ProgrammingError,
255
257
                        "asynchronous query already in execution");
256
258
        return NULL;
257
259
    }
 
260
    pthread_mutex_unlock(&(self->conn->lock));
258
261
    
259
262
    if (PyUnicode_Check(operation)) {
260
263
        PyObject *enc = PyDict_GetItemString(psycoEncodings,
498
501
    
499
502
    /* check if the fetching cursor is the one that did the asynchronous query
500
503
       and raise an exception if not */
 
504
    pthread_mutex_lock(&(self->conn->lock));
501
505
    if (self->conn->async_cursor != NULL
502
506
        && self->conn->async_cursor != (PyObject*)self) {
 
507
        pthread_mutex_unlock(&(self->conn->lock));
503
508
        PyErr_SetString(ProgrammingError,
504
509
                        "asynchronous fetch by wrong cursor");
505
510
        return -2;
506
511
    }
 
512
    pthread_mutex_unlock(&(self->conn->lock));
507
513
    
508
514
    if (self->pgres == NULL) {
509
515
        Dprintf("_psyco_curs_prefetch: trying to fetch data");
622
628
    /* if the query was async aggresively free pgres, to allow
623
629
       successive requests to reallocate it */
624
630
    if (self->row >= self->rowcount
625
 
        
626
631
        && self->conn->async_cursor == (PyObject*)self)
627
632
        IFCLEARPGRES(self->pgres);
628
 
    
 
633
 
629
634
    return res;
630
635
}
631
636
 
687
692
 
688
693
    /* if the query was async aggresively free pgres, to allow
689
694
       successive requests to reallocate it */
690
 
    if (self->row >= self->rowcount && self->conn->async_cursor)
 
695
    if (self->row >= self->rowcount
 
696
        && self->conn->async_cursor == (PyObject*)self)
691
697
        IFCLEARPGRES(self->pgres);
692
698
    
693
699
    return list;
743
749
 
744
750
    /* if the query was async aggresively free pgres, to allow
745
751
       successive requests to reallocate it */
746
 
    if (self->row >= self->rowcount && self->conn->async_cursor)
 
752
    if (self->row >= self->rowcount
 
753
        && self->conn->async_cursor == (PyObject*)self)
747
754
        IFCLEARPGRES(self->pgres);
748
755
    
749
756
    return list;
929
936
static PyObject *
930
937
psyco_curs_fileno(cursorObject *self, PyObject *args)
931
938
{
 
939
    long int socket;
 
940
    
932
941
    if (!PyArg_ParseTuple(args, "")) return NULL;
933
942
    EXC_IF_CURS_CLOSED(self);
934
943
 
935
944
    /* note how we call PQflush() to make sure the user will use
936
945
       select() in the safe way! */
 
946
    pthread_mutex_lock(&(self->conn->lock));
 
947
    Py_BEGIN_ALLOW_THREADS;
937
948
    PQflush(self->conn->pgconn);
938
 
    return PyInt_FromLong((long int)PQsocket(self->conn->pgconn));
 
949
    socket = (long int)PQsocket(self->conn->pgconn);
 
950
    Py_END_ALLOW_THREADS;
 
951
    pthread_mutex_unlock(&(self->conn->lock));
 
952
 
 
953
    return PyInt_FromLong(socket);
939
954
}
940
955
 
941
956
/* extension: isready - return true if data from async execute is ready */
949
964
    if (!PyArg_ParseTuple(args, "")) return NULL;
950
965
    EXC_IF_CURS_CLOSED(self);
951
966
 
 
967
    /* pq_is_busy does its own locking, we don't need anything special but if
 
968
       the cursor is ready we need to fetch the result and free the connection
 
969
       for the next query. */
 
970
    
952
971
    if (pq_is_busy(self->conn)) {
953
972
        Py_INCREF(Py_False);
954
973
        return Py_False;
955
974
    }
956
975
    else {
 
976
        IFCLEARPGRES(self->pgres);
 
977
        pthread_mutex_lock(&(self->conn->lock));
 
978
        self->pgres = PQgetResult(self->conn->pgconn);
 
979
        self->conn->async_cursor = NULL;
 
980
        pthread_mutex_unlock(&(self->conn->lock));
957
981
        Py_INCREF(Py_True);
958
982
        return Py_True;
959
983
    }
1061
1085
            self, ((PyObject *)self)->ob_refcnt);
1062
1086
    
1063
1087
    self->conn = conn;
 
1088
    Py_INCREF((PyObject*)self->conn);
 
1089
    
1064
1090
    self->closed = 0;
1065
1091
    
1066
1092
    self->pgres = NULL; 
1092
1118
{
1093
1119
    cursorObject *self = (cursorObject *)obj;
1094
1120
 
1095
 
    /* if necessary remove cursor from connection */
1096
 
    if (self->conn != NULL) {
1097
 
        PyObject *t;
1098
 
        int len, i;
1099
 
 
1100
 
        if ((len = PyList_Size(self->conn->cursors)) > 0) {
1101
 
            for (i = 0; i < len; i++) {
1102
 
                t = PyList_GET_ITEM(self->conn->cursors, i);
1103
 
                if (self == (cursorObject *)t) {
1104
 
                    Dprintf("cursor_dealloc: found myself in cursor list");
1105
 
                    PySequence_DelItem(self->conn->cursors, i);
1106
 
                    break;
1107
 
                }
1108
 
            }
1109
 
        }
1110
 
    }
1111
1121
 
1112
1122
    if (self->query) free(self->query);
1113
 
    
 
1123
 
 
1124
    Py_DECREF((PyObject*)self->conn);
1114
1125
    Py_XDECREF(self->casts);
1115
1126
    Py_XDECREF(self->description);
1116
1127
    Py_XDECREF(self->pgstatus);