`write_row` busy loop

[jerch]

Found some weird behavior in version 3.0.10 (in both C and python variant, installed from pip).

This seems never to return, it sticks at 100% python CPU with no postgres activity:

def provoke(conn):
    txt = '\\' * 536870883
    c = conn.cursor()
    c.execute('create temporary table ttt(t1 text, t2 text, t3 text, t4 text, t5 text)')
    with c.copy('COPY ttt (t1,t2) FROM STDIN') as copy:
        copy.write_row([txt] * 2)
    c.execute('select length(t1), length(t2), length(t3) from ttt')
    print(c.fetchall())
    c.execute('drop table ttt')
provoke(connection)

Have not yet further debugged, maybe some sort of busy loop in text escaping solely on python side. The value size was chosen for typical postgres limits, which should raise here, but it never gets to postgres side of things.

[jerch]

Some progress with the python version:

Initially it hangs on the \\ --> \\\\ substitution done here:

psycopg/psycopg/psycopg/copy.py

Line 562 in 4d5698f

out += _dump_re.sub(_dump_sub, b)

After waiting for like 5min it quits there with a memory error. To get around the regexp memory error, I temporarily replaced the replacement function with this one (which no full replacement yet):

cheap_repl = lambda v : v.replace(b'\\', b'\\\\')

which gets through the data somewhat faster without a memory error. But now things hang here:

psycopg/psycopg/psycopg/pq/pq_ctypes.py

Line 577 in 4d5698f

rv = impl.PQputCopyData(self._pgconn_ptr, buffer, len(buffer))

thus I aborted it manually after 10 mins. Looking at the postgres docs for that function, it states:

The application can divide the COPY data stream into buffer loads of any convenient size.

thus I tried a quick replacement with a chunked version:

    def put_copy_data(self, buffer: bytes) -> int:
        # TODO: should be done without copy
        if not isinstance(buffer, bytes):
            buffer = bytes(buffer)
        if len(buffer) > 2**20:
            pos = 0
            while pos < len(buffer):
                part = buffer[pos:pos+(2**20)]
                rv = impl.PQputCopyData(self._pgconn_ptr, part, len(part))
                if rv < 0:
                    raise e.OperationalError(f"sending copy data failed: {error_message(self)}")
                pos += 2**20
        else:
            rv = impl.PQputCopyData(self._pgconn_ptr, buffer, len(buffer))
            if rv < 0:
                raise e.OperationalError(f"sending copy data failed: {error_message(self)}")
        return rv

Now it gets through to the expected error from postgres:

psycopg.errors.ProgramLimitExceeded: out of memory
DETAIL:  Cannot enlarge string buffer containing 1073676288 bytes by 65535 more bytes.
CONTEXT:  COPY ttt, line 1

Not sure yet whats the issue in C, maybe also the PQputCopyData call?

[dvarrazzo]

You could try and take a look via py-spy or some other tool to inspect the C stack to see where is the program stuck. See this article for an example.

If it's stuck client-side, I'm moderately sure it is in some libpq escape function.

Is the behaviour linear or quadratic in the string size?

[jerch]

Thx for the profiling/debugging pointer, I have no experience beside the standard cProfiler and debugging packages, so might need some help to get the cython stuff into perspective.

Was able to track the issue with PQputCopyData further down - it is indeed an infinite loop in:

psycopg/psycopg/psycopg/generators.py

Lines 205 to 208 in 4d5698f

	def copy_to(pgconn: PGconn, buffer: bytes) -> PQGen[None]:
	# Retry enqueuing data until successful
	while pgconn.put_copy_data(buffer) == 0:
	yield Wait.W

caused by these circumstances:

• connection is set to non blocking
• PQputCopyData calls pqCheckOutBufferSpace in line 2660, if that call fails it will return 0 for nonblocking connections
• looking at pqCheckOutBufferSpace it will only fail, if realloc does not succeed, adding an error message and returning EOF

So while the outgoing buffer has a max limit, it is not reflected directly at PQputCopyData in non blocking mode. Maybe there should be a peek into the error messages, if PQputCopyData returned 0?(Nope, this does not solve the issue to spot the max limit overflow from a single bucket) In any case I think it is needed to handle big data in chunks, this also would automatically avoid that buffer limit issue. (Have not yet checked how copy.write is passed down to libpq and whether big file chunks would also be affected.)
Furthermore I'd suggest to slow down the busy waiting loop in generators.copy_to by a tiny sleep, otherwise this may get very CPU hungry for no benefit.

Edit:
Thinking more about it - it seems whats missing here, is a flow control/pressure setup. If I get your remarks about the copy cursor behavior right, it will only sync the worker thread state back to the main thread at the end of the context. But doesnt this open the door for OOM on worker thread side (or in the queue), if tons of data arrive too fast? Maybe you already have something in place that I've overlooked (plz correct me, if my assumptions are wrong)...

[dvarrazzo]

Thank you for your analysis. It's not pretty, but maybe we can look if there's memory in the error message and bail out there.

Furthermore I'd suggest to slow down the busy waiting loop in generators.copy_to by a tiny sleep, otherwise this may get very CPU hungry for no benefit.

Yielding a Wait.W should trigger scheduling. It doesn't?

But doesnt this open the door for OOM on worker thread side (or in the queue), if tons of data arrive too fast? The queue has a limited size and blocks.

The queue has a max size of 1024 items. Every item should be about 32K items. As a result it shouldn't grow more than 32Mb of space.

[dvarrazzo]

Have not yet checked how copy.write is passed down to libpq and whether big file chunks would also be affected

We could add a chunking mechanism downstream, in a code path common to both write() and write_row()

[jerch]

Yielding a Wait.W should trigger scheduling. It doesn't?

The waiting.wait construct on the db socket does not seem to kick in for the toxic data - I guess because it is not blocking at all (as the socket has not seen any payload at this point) and returns immediately? Currently that loop runs at 100% CPU, with most time spent in pqCheckOutBufferSpace (prolly at the 8KB buffer expansion logic). I still wonder where that max limit comes from on libpq side, following realloc includes/defines I cannot find it being bound to postgres limits. (And I kinda dont feel yet like building the lib itself in debug mode 😸)

We could add a chunking mechanism downstream, in a code path common to both write() and write_row().

Sounds like it would solve all the issues at once - PQputCopyData would play along and its invocation can be kept as it is, and also the queue can not grow overly anymore, if single chunks have an upper limit. Downside though - prechunking loses information on the source of the data bucket (no clue if the code relies on that somewhere).

NB: That all for the python version so far, C might be a different story....

[dvarrazzo]

Downside though - prechunking loses information on the source of the data bucket (no clue if the code relies on that somewhere).

I don't think it does, no, so it's on my radar to do so. First I want to play with your example a bit: thank you for the useful repro.

NB: That all for the python version so far, C might be a different story....

I don't think so. C is only concerned with optimising the formatting and functions, not with the IO, and implements the thin wrapper PGconn.put_copy_data() around the PQputCopyData() function.

EDIT - ah no, there is a generator too, but there is nothing that should behave differently between Python and C.

[jerch]

Btw regarding the byte limit:

PQputCopyData(PGconn *conn, const char *buffer, int nbytes)

Why is the byte size typed as int? 😱 (Hopefully this does not continue deeper into pq source, otherwise negative values might cause greater havoc here...)

---

Edit:
Imho PQputCopyData needs some overhaul - currently for an overflowed int value into negative it would return 1 indicating data was sent, but didnt do anything. What? Shouldn't it use size_t in the first place, with some explicit limit checks (if there is a lower limit than from C lib/arch).

---

Edit2:
Ok found it, the out buffer limit is indeed limited against MaxAllocSize, which is hardcoded as ~1GB (0x3fffffff). It happens via pqCheckOutBufferSpace --> realloc --> repalloc --> AllocSizeIsValid --> MaxAllocSize symbol/call chain.

[dvarrazzo]

Yes, there are some pretty old part of the libpq which use int instead of size_t...

I have created a branch with fixes for this problem. It relies on finding memory in the error message, so it's not pretty. Will have to test it better trying to trigger a would-block state in put_copy_data().

Also fixed a bug with the working thread: the exception is now passed to the main thread and raised from there. Needs testing and to verify if similar code is necessary in the async path.

Note that the C path of the backslash escaping is much faster and only takes 3 seconds to forward the data to the now-crashing function.

(.venv) piro@baloo:~/dev/psycopg3$ time PSYCOPG_IMPL=c python testcopy.py 
Traceback (most recent call last):
  File "testcopy.py", line 18, in <module>
    provoke(conn)
  File "testcopy.py", line 9, in provoke
    copy.write_row([txt] * 2)
  File "/usr/lib/python3.8/contextlib.py", line 120, in __exit__
    next(self.gen)
  File "/home/piro/dev/psycopg3/psycopg/psycopg/cursor.py", line 683, in copy
    yield copy
  File "/home/piro/dev/psycopg3/psycopg/psycopg/copy.py", line 197, in __exit__
    self.finish(exc_val)
  File "/home/piro/dev/psycopg3/psycopg/psycopg/copy.py", line 264, in finish
    self._write_end()
  File "/home/piro/dev/psycopg3/psycopg/psycopg/copy.py", line 311, in _write_end
    raise self._worker_error
  File "/home/piro/dev/psycopg3/psycopg/psycopg/copy.py", line 283, in worker
    self.connection.wait(copy_to(self._pgconn, data))
  File "/home/piro/dev/psycopg3/psycopg/psycopg/connection.py", line 861, in wait
    return waiting.wait(gen, self.pgconn.socket, timeout=timeout)
  File "/home/piro/dev/psycopg3/psycopg/psycopg/waiting.py", line 222, in wait_epoll
    s = next(gen)
  File "/home/piro/dev/psycopg3/psycopg/psycopg/generators.py", line 216, in copy_to
    raise e.OperationalError(msg)
psycopg.OperationalError: cannot allocate memory for output buffer


real	0m3.265s
user	0m2.583s
sys	0m0.672s

How does it look?

[jerch]

How does it look?

Very nice 👍.

But will the memory error message scan be reliable to distinct slow transport processing (old buckets stacking up in out buffer) vs. a single bucket overflowing the max buffer limit? From looking at pqCheckOutBufferSpace and pqPutnchar --> pqPutMsgBytes it seems, new buckets simply get appended to the out buffer, so pressure from slow connection also might run into the same memory error?

Edit:
Btw there is a double call perf smell there - pqCheckOutBufferSpace gets called twice down the PQputCopyData path, with slightly different buffer needs though (with only additional constant bytes for PQputCopyData, thus could be merged into only one more clever call). Ah - I should stop nagging about libpq...

[dvarrazzo]

new buckets simply get appended to the out buffer, so pressure from slow connection also might run into the same memory error?

Do you mean the buffer to compose the error message? That would be a small realloc, not a multiMB realloc. I sincerely hope the OOM killer will come and put end to the misery of this process 💀

Btw there is a double call perf smell there

but likely there will be only one re-allocation, the other would hit the happy path and bail out early I expect, no?

Ah - I should stop nagging about libpq...

if you have observations about the libpq code you may get in touch with the postgres devs at pgsql-hackers mailing lists. Observations about what makes the life of a libpq client harder may be welcome. Even this conflating a memory error with what the docs suggest is a would-block condition smells dodgy.

[jerch]

There might be another perf smell hidden in libpq, if used with non blocking connection - in pqSendSome at line 943 the whole buffer gets realigned for every successfully sent portion. I understand, that this might be needed for complicated SQL stuff (guess they use that for all client->server comm?), but in case of big data from COPY FROM it creates alot of unnecessary memory juggling, where a small ringbuffer abstraction would be like 10 times faster -.-

Do you mean the buffer to compose the error message? That would be a small realloc, not a multiMB realloc. I sincerely hope the OOM killer will come and put end to the misery of this process.

Ah nope, I mean the libpq out buffer, where it stores everything to be sent to the server. Currently I am not quite sure about the mechanics here for non blocking connections, will see, if I can mockup a test case to repro, what I currently think it does.

Even this conflating a memory error with what the docs suggest is a would-block condition smells dodgy.

Yes, it could at least have set errno, so a custom client binding can check for rare irregularities (or even easier - by another fault state return value).

[dvarrazzo]

pqCheckOutBufferSpace() is an internal function so I think they can establish a return error to report an error with palloc(), which might be assumed not recoverable.

After securing psycopg we can drop a line to the -hackers to hear what they think and if there's space for improvement there.